package com.siteseat.crudeq.broker.netty;

import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.ChannelBuffers;
import org.jboss.netty.channel.Channel;

import com.siteseat.common.security.SecurityManager;
import com.siteseat.crudeq.broker.transport.impl.BaseBrokeSession;
import com.siteseat.crudeq.common.raw.protocol.v1.V1ResponseId;
import com.siteseat.crudeq.kernel.Chunk;
import com.siteseat.crudeq.kernel.ChunkInfo;
import com.siteseat.crudeq.kernel.Message;
import com.siteseat.crudeq.kernel.QueueManager;

/**
 *
 */
public class NettyBrokerSession extends BaseBrokeSession {

  private Channel channel;

  /**
   * @param internalId
   */
  public NettyBrokerSession( Channel channel, SecurityManager authenticator, QueueManager queueManager ) {
    super( authenticator, queueManager );

    assert channel.isOpen();
    this.channel = channel;
  }

  /**
   * {@inheritDoc}
   */
  @Override
  protected void sendMessage( Message msg ) {
    ChannelBuffer output = ChannelBuffers.dynamicBuffer();

    Chunk c = msg.getChunk();
    ChunkInfo ci = c.getInfo();
    int msgSize = msg.getSize();
    int offset = msg.getRawOffset();
    
    output.writeByte( V1ResponseId.MESSAGE );
    output.writeInt( 4 + 8 + 4 + msgSize );
    
    output.writeInt( ci.getQueueId().intValue() );
    output.writeLong( ci.getChunkId().longValue() );
    output.writeInt( offset );
    output.writeBytes( c.getPayload(), offset, msgSize );
    
    channel.write( output );
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public Integer getInternalId() {
    return getChannel().getId();
  }

  /**
   * Gets the channel
   * 
   * @return the channel
   */
  public Channel getChannel() {
    return channel;
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public void close() {
    if( this.channel.isOpen() ) {
      this.channel.close();
    }
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public boolean isConnected() {
    return this.channel.isConnected();
  }

}
